/**
 * 
 */
package com.cqcis.uip.base.framework.utils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.util.NamePreservingRunnable;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.core.LogService;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.utils.LogServiceThreadFactory.LogThread;
import com.cqcis.uip.base.logadaptor.LogConfiguration;

/**
 * 日志服务线程池, 平台中的所有线程产生都必须用该线程池产生。
 * 不推荐自定义线程池。
 * 
 * @author huangmy
 *
 */
public class LogServiceThreadPoolExecutor {	
	private final static Log log = LogFactory.getLog(LogServiceThreadPoolExecutor.class);
	private final static AtomicLong THREAD_POOL_ID = new AtomicLong(0);
	private final static int CORE_SIZE_DEFAULT = 12; // 默认的核心线程数
	private final static int MAX_SIZE_DEFAULT = CORE_SIZE_DEFAULT * 2; // 默认的最大线程数
	private final static int KEEPALIVE_TIME_DEFAULT = 60; // 默认存活时间
	// 线程池与线程对应关系维护
	private final static Map<ThreadPoolExecutor, Map<Runnable, Thread>> tpeMaps = new ConcurrentHashMap<ThreadPoolExecutor, Map<Runnable,Thread>>();
	
	private LogServiceThreadPoolExecutor() {		
	}
	
	/**
	 * 忽略线程名称, 线程名称采用默认方式
	 * 
	 * @param coreSize
	 * @param maxSize
	 * @param keepAliveTime
	 * @param logService
	 * @param logConfiguration
	 * 	 
	 * @return
	 */
	public static ThreadPoolExecutor getLogExecutor(int coreSize, int maxSize, int keepAliveTime, 
            final LogService logService, final LogConfiguration logConfiguration) {
		return getLogExecutor(coreSize, maxSize, keepAliveTime, logService, logConfiguration, null, new LinkedBlockingQueue<Runnable>(coreSize * 2));
	}
	
	/**
	 * 忽略线程名称, 线程名称采用默认方式
	 * 定义于RS
	 * 
	 * @param coreSize
	 * @param maxSize
	 * @param keepAliveTime
	 * @param serviceContext
	 * @return
	 */
	public static ThreadPoolExecutor getLogExecutor(int coreSize, int maxSize, int keepAliveTime, RegisteredServiceContext serviceContext) {
		return getLogExecutor(coreSize, maxSize, keepAliveTime, serviceContext.getLogService(), getLogConfiguration(serviceContext));
	}
	
	private static LogConfiguration getLogConfiguration(RegisteredServiceContext serviceContext) {
		RegisteredService rs = serviceContext.getRegisteredService();
		if (rs != null) {
			return rs.getLogConfiguration();
		}
		return null;
	}
	
	protected static String getLogThreadSuffixName(RegisteredServiceContext serviceContext) {
		RegisteredService rs = serviceContext.getRegisteredService();
		if (rs != null) {
			return rs.getName();
		}
		return "";
	}
	
	/**
	 * 忽略线程名称, 线程名称采用默认方式
	 * 
	 * @param coreSize
	 * @param maxSize
	 * @param keepAliveTime
	 * @param logService
	 * @param logConfiguration
	 * @param blockingRunnableQueue
	 * 	 
	 * @return
	 */
	public static ThreadPoolExecutor getLogExecutor(int coreSize, int maxSize, int keepAliveTime, 
            final LogService logService, final LogConfiguration logConfiguration, BlockingQueue<Runnable> blockingRunnableQueue) {
		return getLogExecutor(coreSize, maxSize, keepAliveTime, logService, logConfiguration, null, blockingRunnableQueue);
	}
	
	/**
	 * 忽略线程名称, 线程名称采用默认方式
	 * 
	 * @param coreSize
	 * @param maxSize
	 * @param keepAliveTime
	 * @param serviceContext
	 * @param blockingRunnableQueue
	 * @return
	 */
	public static ThreadPoolExecutor getLogExecutor(int coreSize, int maxSize, int keepAliveTime, 
			RegisteredServiceContext serviceContext, BlockingQueue<Runnable> blockingRunnableQueue) {
		return getLogExecutor(coreSize, maxSize, keepAliveTime, serviceContext.getLogService(), 
				getLogConfiguration(serviceContext), blockingRunnableQueue);
	}
	
	/**
	 * 获取带日志服务的线程池
	 * 
	 * @param coreSize
	 * @param maxSize
	 * @param keepAliveTime
	 * @param logService
	 * @param logdirname
	 * @param logConfiguration
	 * @param threadPoolName
	 * 
	 * @return
	 */
	public static LogThreadPoolExecutor getLogExecutor(int coreSize, int maxSize, int keepAliveTime, 
			            final LogService logService, final LogConfiguration logConfiguration, final String threadPoolName) {
		coreSize = coreSize <= 0 ? CORE_SIZE_DEFAULT : coreSize;
		
		return getLogExecutor(coreSize, maxSize, keepAliveTime, logService, logConfiguration, threadPoolName, 
				new LinkedBlockingQueue<Runnable>(coreSize * 2));//2倍核心线程数
	}
	
	/**
	 * 获取带日志服务的线程池
	 * 
	 * @param coreSize
	 * @param maxSize
	 * @param keepAliveTime
	 * @param serviceContext
	 * @param threadPoolName
	 * @return
	 */
	public static ThreadPoolExecutor getLogExecutor(int coreSize, int maxSize, int keepAliveTime, 
			RegisteredServiceContext serviceContext, final String threadPoolName) {
		return getLogExecutor(coreSize, maxSize, keepAliveTime, serviceContext.getLogService(), 
				getLogConfiguration(serviceContext), threadPoolName);
	}
	
	/**
	 * 获取带日志服务的线程池，提供不同的阻塞队列方式
	 * 
	 * @param coreSize
	 * @param maxSize
	 * @param keepAliveTime
	 * @param serviceContext
	 * @param threadPoolName
	 * @param blockingRunnableQueue
	 * @return
	 */
	public static ThreadPoolExecutor getLogExecutor(int coreSize, int maxSize, int keepAliveTime, 
			RegisteredServiceContext serviceContext, final String threadPoolName, BlockingQueue<Runnable> blockingRunnableQueue) {
		return getLogExecutor(coreSize, maxSize, keepAliveTime, serviceContext.getLogService(), 
				getLogConfiguration(serviceContext), threadPoolName, blockingRunnableQueue);
	}
	
	/**
	 * 获取带日志服务的线程池，提供不同的阻塞队列方式
	 * 
	 * @param coreSize
	 * @param maxSize
	 * @param keepAliveTime
	 * @param logService
	 * @param logdirname
	 * @param logConfiguration
	 * @param threadPoolName
	 * @param blockingRunnableQueue
	 * 
	 * @return
	 */
	public static LogThreadPoolExecutor getLogExecutor(int coreSize, int maxSize, int keepAliveTime, 
			            final LogService logService, final LogConfiguration logConfiguration, final String threadPoolName, BlockingQueue<Runnable> blockingRunnableQueue) {
		coreSize = coreSize <= 0 ? CORE_SIZE_DEFAULT : coreSize;
		maxSize = maxSize <= 0 ? MAX_SIZE_DEFAULT : maxSize;
		keepAliveTime = keepAliveTime <= 0 ? KEEPALIVE_TIME_DEFAULT : keepAliveTime;
		
		maxSize = maxSize >= coreSize ? maxSize : coreSize;
		
		final ConcurrentTaskQueue<Runnable> rejectRunnableQueue = new ConcurrentTaskQueue<Runnable>();
		LogThreadPoolExecutor tpe = new LogThreadPoolExecutor(coreSize, maxSize, 
				keepAliveTime, TimeUnit.SECONDS,
				blockingRunnableQueue,// 一般为2倍核心线程数
	            new LogServiceThreadFactory(),
	            new RejectedExecutionHandler() { // 被拒绝
			public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
				rejectRunnableQueue.offer(r);				
			}
		}) {
			@Override
			public void shutdown() {
				try {
					super.shutdown();
				    shutdownNow();
				} catch (Throwable t) {		
					log.error("Catch Exception in shutdown.", t);
				}
			}
			
			@Override
			public List<Runnable> shutdownNow() {
				List<Runnable> runs = null;
				try {
					runs = super.shutdownNow();
				    tpeMaps.remove(this);
				    runs = super.shutdownNow();
				} catch (Throwable t) {			
					log.error("Catch Exception in shutdown.", t);
				}
				
				return runs;
			}
			
			@Override
			public void execute(final Runnable command) {
				Runnable cmdRund = new Runnable() {
					public void run() {
						command.run(); // 执行传入任务
						
						if (!isShutdown()) { // 线程池没有被关闭
		        			// 继续运行前面被拒绝的线程任务
		            		Runnable rejectRun = null;
		            		while ((rejectRun = rejectRunnableQueue.remove()) != null) {
		            			rejectRun.run();
		            		}
		        		}
					}
				};
				
				super.execute(cmdRund); // 调用父类
			}
		};
				
		tpeMaps.put(tpe, new ConcurrentHashMap<Runnable, Thread>());
		
	    tpe.setThreadFactory(createLogThreadFactory(tpe, rejectRunnableQueue, logService, logConfiguration, threadPoolName));
	    
	    return tpe;
	}
	
	/**
	 * 保存线程池到日志配置中
	 * 
	 * @param logConfiguration
	 * @param tpe
	 */
	private static void saveTPEforLogConfig(LogConfiguration logConfiguration, ThreadPoolExecutor tpe) {
		if (logConfiguration != null && tpe != null) {
			LogConfigurationUtil.registerTPEFor(logConfiguration, tpe);
		}		
	}
	
	/**
	 * 忽略线程名称
	 * 
	 * @param tpe
	 * @param logService
	 * @param logConfiguration
	 * @return
	 */
	public static ThreadFactory createLogThreadFactory(ThreadPoolExecutor tpe, 
			ConcurrentTaskQueue<Runnable> rejectRunnableQueue, LogService logService, LogConfiguration logConfiguration) {
		return createLogThreadFactory(tpe, rejectRunnableQueue, logService, logConfiguration, null);
	}
	
	/**
	 * 生成指定线程池的日志服务线程工厂
	 * 
	 * @param tpe
	 * @param logService
	 * @param logConfiguration
	 * @return
	 */
	private static ThreadFactory createLogThreadFactory(final ThreadPoolExecutor tpe,
			     final ConcurrentTaskQueue<Runnable> rejectRunnableQueue,
			     final LogService logService, final LogConfiguration logConfiguration, final String threadName) {
		final ThreadFactory originalThreadFactory = tpe.getThreadFactory();
		final long threadPoolId = THREAD_POOL_ID.incrementAndGet();
		final String poolName = (threadName == null ? FrameworkConstants.SERVICE_THREAD_NAME_PREFIX : threadName);// 线程池名称
        if (tpe instanceof LogThreadPoolExecutor) {
        	LogThreadPoolExecutor logTpe = (LogThreadPoolExecutor)tpe;
        	logTpe.setId(String.valueOf(threadPoolId)); // 设置线程池ID
        	logTpe.setName(poolName);// 线程池名称
        }
                
	    ThreadFactory newThreadFactory = new ThreadFactory() {
	        private final AtomicLong threadId = new AtomicLong(0);
	        	        	        
	        public Thread newThread(final Runnable runnable) {
	        	Runnable runnto = new Runnable() {
                	public void run() {
                		if (logService != null && logConfiguration != null) {
                			logService.register(logConfiguration);
                			Collection<LogConfiguration> relaConfigs = logConfiguration.getAllRelationConfigs();
                			if (relaConfigs != null) {
                				for (LogConfiguration relaConfig : relaConfigs) {
                					logService.register(relaConfig);
                				}
                			}
                		}
                		
                		runnable.run();// 运行传入的线程任务
                		
                		if (!tpe.isShutdown()) { // 线程池没有被关闭
                			// 继续运行前面被拒绝的线程任务
                    		Runnable rejectRun = null;
                    		while ((rejectRun = rejectRunnableQueue.remove()) != null) {
                    			rejectRun.run();
                    		}
                		}
                		                		
                		Map<Runnable, Thread> rtms = tpeMaps.get(tpe);
                		if (rtms != null) {
                			rtms.remove(this);
                		}
                		
                		if (logService != null && logConfiguration != null) {
                			logService.unregister();
                		}
                	}
                };
	        	
                final long tId = threadId.incrementAndGet();
	            final Thread t = originalThreadFactory.newThread(
	                    new NamePreservingRunnable(
	                    		runnto, 
	                    		poolName + "-P" + threadPoolId + "-" + tId));
	            t.setDaemon(true);
	            LogThread logthread = (LogThread)t;
	            logthread.setLogThreadId(threadPoolId + ":" + tId);
	            
	            Map<Runnable, Thread> rtms = tpeMaps.get(tpe);
        		if (rtms != null) {
        			rtms.put(runnto, t);
        		}
	            
	            return t;
	        }
	    };
	    
	    saveTPEforLogConfig(logConfiguration, tpe);
	    
	    return newThreadFactory;
	}
	
	/**
	 * 获取指定线程池中的所有活动线程, 如果指定线程池不是由该线程池工具产生的，则返回null
	 * 
	 * @param tpe
	 * @return
	 */
	public static Collection<Thread> getAllActiveThreadsFrom(ThreadPoolExecutor tpe) {
		Map<Runnable, Thread> rtms = tpeMaps.get(tpe);
		if (rtms != null) {
			List<Thread> threads = new ArrayList<Thread>();
			for (Iterator<Thread> it = rtms.values().iterator(); it.hasNext();) {
				threads.add(it.next());
			}
			
			return threads;
		}
		
		return null;
	}
	
	/**
	 * 获取当前的线程池
	 * 
	 * @param id
	 * @return
	 */
	public static ThreadPoolExecutor getCurrentThreadPoolExecutorById(String id) {
		for (ThreadPoolExecutor tpe : tpeMaps.keySet()) {
			if (tpe instanceof LogThreadPoolExecutor) {
				LogThreadPoolExecutor logTpe = (LogThreadPoolExecutor)tpe;
				if (logTpe.getId().equals(id)) {
					return tpe;
				}
			}
		}
		return null;
	}
	
	/**
	 * 阻塞队列
	 * 
	 * @param size
	 * @return
	 */
	public static BlockingQueue<Runnable> getBlockQueue(int size) {
		BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(size) {			
			private static final long serialVersionUID = 0L;

			@Override
			public boolean offer(Runnable o) {
				try {
					super.put(o);
				} catch (InterruptedException e) {
					return false;
				}
				return true;
			}
		};
		
		return queue;
	}
	
	public static class LogThreadPoolExecutor extends ThreadPoolExecutor {
		private Map<String, Object> attributes = new ConcurrentHashMap<String, Object>();// 属性
		private String name; // 名称
		private String id; // ID
		private CountDownLatch waitShutdownLatch = new CountDownLatch(2);
		
		public LogThreadPoolExecutor(int corePoolSize,
                int maximumPoolSize,
                long keepAliveTime,
                TimeUnit unit,
                BlockingQueue<Runnable> workQueue) {
			super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
		}
		
		public LogThreadPoolExecutor(int corePoolSize,
                int maximumPoolSize,
                long keepAliveTime,
                TimeUnit unit,
                BlockingQueue<Runnable> workQueue,
                ThreadFactory threadFactory) {
			super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
		}
		
		public LogThreadPoolExecutor(int corePoolSize,
                int maximumPoolSize,
                long keepAliveTime,
                TimeUnit unit,
                BlockingQueue<Runnable> workQueue,
                RejectedExecutionHandler handler) {
			super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
		}
		
		public LogThreadPoolExecutor(int corePoolSize,
                int maximumPoolSize,
                long keepAliveTime,
                TimeUnit unit,
                BlockingQueue<Runnable> workQueue,
                ThreadFactory threadFactory,
                RejectedExecutionHandler handler) {
			super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
		}
		
		public void shutdownWait() {
			shutdownWait(0);
		}
		
		public void shutdownWait(long waittime) {
			super.shutdown();	
			
			this.waitShutdownLatch.countDown();
			
			super.shutdown();
			
			try {
				if (waittime <= 0) {
					this.waitShutdownLatch.await();
				} else {
					this.waitShutdownLatch.await(waittime, TimeUnit.MILLISECONDS);
				}
			} catch (InterruptedException e) {
			}
			
			shutdown();
			
			super.shutdown();
			shutdown();
		}
		
		@Override
		protected void terminated() {
			this.waitShutdownLatch.countDown();
			super.terminated();
		}
		
		public Map<String, Object> getAttributes() {
			return attributes;
		}

		public void setAttributes(Map<String, Object> attributes) {		
			this.attributes = attributes;
		}
	    	
		public Object setAttribute(String name, Object value) {		
			return this.attributes.put(name, value);
		}
		
		public Object getAttribute(String name) {
	        return this.attributes.get(name);
		}
		
		public Object removeAttribute(String name) {		
			return this.attributes.remove(name);
		}

		/**
		 * 获取名称
		 * 
		 * @return
		 */
		public String getName() {
			return name;
		}

		/**
		 * 设置名称
		 * 
		 * @param name
		 */
		public void setName(String name) {
			this.name = name;
		}

		public String getId() {
			return id;
		}

		public void setId(String id) {
			this.id = id;
		}
	}
}
